package com.seedog.mq.distributed;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 集群管理器实现类，提供分布式集群管理功能
 */
public class ClusterManagerImpl implements ClusterManager {
    private static final Logger logger = LoggerFactory.getLogger(ClusterManagerImpl.class);
    
    private DistributedConfig config;
    private Node currentNode;
    private final Map<String, Node> nodes = new ConcurrentHashMap<>();
    private final AtomicReference<Node> masterNode = new AtomicReference<>();
    private final Map<String, MessageHandler> messageHandlers = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3);
    private final ExecutorService messageProcessor = Executors.newCachedThreadPool();
    private final AtomicBoolean running = new AtomicBoolean(false);
    private final BlockingQueue<ClusterMessage> messageQueue = new LinkedBlockingQueue<>();
    
    public ClusterManagerImpl() {
        // 初始化消息处理器线程
        startMessageProcessor();
    }
    
    @Override
    public void initialize(DistributedConfig config) {
        if (config == null) {
            throw new IllegalArgumentException("DistributedConfig cannot be null");
        }
        
        this.config = config;
        this.currentNode = new Node(config.getHost(), config.getPort());
        this.currentNode.setStatus(Node.NodeStatus.INACTIVE);
        nodes.put(currentNode.getNodeId(), currentNode);
        
        logger.info("ClusterManager initialized with current node: {}", currentNode);
    }
    
    @Override
    public List<Node> getAllNodes() {
        return new ArrayList<>(nodes.values());
    }
    
    @Override
    public List<Node> getActiveNodes() {
        List<Node> activeNodes = new ArrayList<>();
        long currentTime = System.currentTimeMillis();
        long timeout = config.getHeartbeatTimeout();
        
        for (Node node : nodes.values()) {
            if (node.isAlive(timeout)) {
                activeNodes.add(node);
            }
        }
        
        return activeNodes;
    }
    
    @Override
    public Node getMasterNode() {
        return masterNode.get();
    }
    
    @Override
    public Node getCurrentNode() {
        return currentNode;
    }
    
    @Override
    public boolean isMaster() {
        Node master = masterNode.get();
        return master != null && master.getNodeId().equals(currentNode.getNodeId());
    }
    
    @Override
    public boolean joinCluster(List<String> seedNodes) {
        checkRunning();
        
        // 简化实现：在实际分布式系统中，这里应该通过网络通信连接到种子节点
        logger.info("Attempting to join cluster with seed nodes: {}", seedNodes);
        
        // 模拟加入集群成功
        currentNode.setStatus(Node.NodeStatus.ACTIVE);
        
        // 如果是第一个节点，自动成为主节点
        if (nodes.size() == 1 && config.isAutoElection()) {
            electMaster();
        }
        
        logger.info("Joined cluster successfully");
        return true;
    }
    
    @Override
    public boolean leaveCluster() {
        if (!running.get()) {
            return true;
        }
        
        logger.info("Leaving cluster...");
        currentNode.setStatus(Node.NodeStatus.SHUTTING_DOWN);
        
        // 如果当前节点是主节点，触发重新选举
        if (isMaster()) {
            triggerReelection();
        }
        
        currentNode.setStatus(Node.NodeStatus.INACTIVE);
        logger.info("Left cluster successfully");
        return true;
    }
    
    @Override
    public void broadcastMessage(Object message) {
        checkRunning();
        
        if (message == null) {
            return;
        }
        
        // 简化实现：在实际分布式系统中，这里应该通过网络通信将消息发送到所有节点
        logger.debug("Broadcasting message: {}", message);
        
        // 模拟广播消息
        for (Node node : getActiveNodes()) {
            if (!node.getNodeId().equals(currentNode.getNodeId())) {
                // 模拟发送到其他节点
                logger.trace("Sending message to node: {}", node.getNodeId());
            }
        }
    }
    
    @Override
    public boolean sendMessageToNode(String nodeId, Object message) {
        checkRunning();
        
        if (nodeId == null || message == null) {
            return false;
        }
        
        Node node = nodes.get(nodeId);
        if (node == null || !node.isAlive(config.getHeartbeatTimeout())) {
            logger.warn("Node {} not found or not alive", nodeId);
            return false;
        }
        
        // 简化实现：在实际分布式系统中，这里应该通过网络通信将消息发送到指定节点
        logger.debug("Sending message to node {}: {}", nodeId, message);
        
        // 模拟发送消息
        return true;
    }
    
    @Override
    public void registerMessageHandler(String messageType, MessageHandler messageHandler) {
        if (messageType == null || messageHandler == null) {
            throw new IllegalArgumentException("Message type and handler cannot be null");
        }
        
        messageHandlers.put(messageType, messageHandler);
        logger.debug("Registered message handler for type: {}", messageType);
    }
    
    @Override
    public void start() {
        if (running.compareAndSet(false, true)) {
            logger.info("Starting ClusterManager...");
            
            // 启动心跳发送任务
            scheduler.scheduleAtFixedRate(this::sendHeartbeat, 
                    config.getHeartbeatInterval(), 
                    config.getHeartbeatInterval(), 
                    TimeUnit.MILLISECONDS);
            
            // 启动节点状态检查任务
            scheduler.scheduleAtFixedRate(this::checkNodeStatus, 
                    config.getHeartbeatTimeout(), 
                    config.getHeartbeatTimeout(), 
                    TimeUnit.MILLISECONDS);
            
            // 启动主节点监控任务（如果启用了自动选举）
            if (config.isAutoElection()) {
                scheduler.scheduleAtFixedRate(this::monitorMaster, 
                        config.getElectionTimeout() / 2, 
                        config.getElectionTimeout() / 2, 
                        TimeUnit.MILLISECONDS);
            }
            
            currentNode.setStatus(Node.NodeStatus.ACTIVE);
            logger.info("ClusterManager started successfully");
        }
    }
    
    @Override
    public void stop() {
        if (running.compareAndSet(true, false)) {
            logger.info("Stopping ClusterManager...");
            
            leaveCluster();
            scheduler.shutdown();
            messageProcessor.shutdown();
            
            try {
                if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                    scheduler.shutdownNow();
                }
                if (!messageProcessor.awaitTermination(5, TimeUnit.SECONDS)) {
                    messageProcessor.shutdownNow();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                scheduler.shutdownNow();
                messageProcessor.shutdownNow();
            }
            
            currentNode.setStatus(Node.NodeStatus.INACTIVE);
            logger.info("ClusterManager stopped successfully");
        }
    }
    
    // 发送心跳
    private void sendHeartbeat() {
        currentNode.updateHeartbeat();
        logger.trace("Sent heartbeat from node: {}", currentNode.getNodeId());
    }
    
    // 检查节点状态
    private void checkNodeStatus() {
        long currentTime = System.currentTimeMillis();
        long timeout = config.getHeartbeatTimeout();
        
        for (Iterator<Map.Entry<String, Node>> it = nodes.entrySet().iterator(); it.hasNext();) {
            Map.Entry<String, Node> entry = it.next();
            Node node = entry.getValue();
            
            // 跳过当前节点
            if (node.getNodeId().equals(currentNode.getNodeId())) {
                continue;
            }
            
            // 检查节点是否超时
            if (!node.isAlive(timeout)) {
                logger.warn("Node {} is timeout, marking as failed", node.getNodeId());
                node.setStatus(Node.NodeStatus.FAILED);
                
                // 如果节点是主节点，触发重新选举
                if (masterNode.get() != null && node.getNodeId().equals(masterNode.get().getNodeId())) {
                    triggerReelection();
                }
            }
        }
    }
    
    // 监控主节点
    private void monitorMaster() {
        Node master = masterNode.get();
        
        // 如果没有主节点或者主节点已经失效，触发选举
        if (master == null || !master.isAlive(config.getHeartbeatTimeout())) {
            electMaster();
        }
    }
    
    // 选举主节点
    private void electMaster() {
        logger.info("Starting master election...");
        
        List<Node> activeNodes = getActiveNodes();
        if (activeNodes.isEmpty()) {
            logger.warn("No active nodes found for election");
            return;
        }
        
        // 简化的选举算法：选择优先级最高、节点ID最小的节点作为主节点
        Node electedMaster = activeNodes.stream()
                .max(Comparator.comparingInt(Node::getPriority)
                        .thenComparing(Node::getNodeId))
                .orElse(null);
        
        if (electedMaster != null) {
            masterNode.set(electedMaster);
            electedMaster.setRole(Node.NodeRole.MASTER);
            
            // 其他节点设为从节点
            for (Node node : activeNodes) {
                if (!node.getNodeId().equals(electedMaster.getNodeId())) {
                    node.setRole(Node.NodeRole.SLAVE);
                }
            }
            
            logger.info("Master elected: {}", electedMaster);
        }
    }
    
    // 触发重新选举
    private void triggerReelection() {
        logger.warn("Triggering reelection due to master node failure");
        masterNode.set(null);
        
        // 延迟触发选举，给其他节点时间处理
        scheduler.schedule(this::electMaster, 1, TimeUnit.SECONDS);
    }
    
    // 启动消息处理器
    private void startMessageProcessor() {
        messageProcessor.submit(() -> {
            while (running.get() || !messageQueue.isEmpty()) {
                try {
                    ClusterMessage message = messageQueue.poll(100, TimeUnit.MILLISECONDS);
                    if (message != null) {
                        processMessage(message);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    logger.error("Error processing message", e);
                }
            }
        });
    }
    
    // 处理消息
    private void processMessage(ClusterMessage message) {
        MessageHandler handler = messageHandlers.get(message.getMessageType());
        if (handler != null) {
            try {
                handler.handleMessage(message.getSenderNodeId(), message.getMessage());
            } catch (Exception e) {
                logger.error("Error handling message of type {}", message.getMessageType(), e);
            }
        } else {
            logger.warn("No handler found for message type: {}", message.getMessageType());
        }
    }
    
    // 检查集群管理器是否正在运行
    private void checkRunning() {
        if (!running.get()) {
            throw new IllegalStateException("ClusterManager is not running");
        }
    }
    
    // 集群消息内部类
    private static class ClusterMessage {
        private final String senderNodeId;
        private final String messageType;
        private final Object message;
        
        public ClusterMessage(String senderNodeId, String messageType, Object message) {
            this.senderNodeId = senderNodeId;
            this.messageType = messageType;
            this.message = message;
        }
        
        public String getSenderNodeId() {
            return senderNodeId;
        }
        
        public String getMessageType() {
            return messageType;
        }
        
        public Object getMessage() {
            return message;
        }
    }
}